#include "client_manager.h"

#include <stdio.h>
#include <netinet/in.h>
#include <unistd.h>


#define ALL_BROADCAST_IP 0xffffffff

static lpmempool_t g_mempool_client = NULL;
static lpmempool_t g_mempool_buffer = NULL;
static lpmempool_t g_mempool_send_req = NULL;
static lpmempool_t g_mempool_send_watcher = NULL;

typedef struct {
	char *buf;
	ssize_t buf_len;
	volatile int rest_cnt;
} send_req_t;

typedef struct {
	struct ev_io watcher;
	client_t *tg_cli;
	send_req_t *req;
} send_watcher_t;

// --------------------------------------------------------

void client_manager_init() {
	g_mempool_client = mempool_init(0, 0, sizeof(client_t));
	g_mempool_buffer = mempool_init(0, 0, BUFFER_SIZE);
	g_mempool_send_req = mempool_init(0, 0, sizeof(send_req_t));
	g_mempool_send_watcher = mempool_init(0, 0, sizeof(send_watcher_t));
}

void client_manager_release() {
	mempool_destroy(&g_mempool_client);
	mempool_destroy(&g_mempool_buffer);
	mempool_destroy(&g_mempool_send_req);
	mempool_destroy(&g_mempool_send_watcher);
}

void all_clients_return(server_params_t *params) {
	pthread_rwlock_wrlock(&params->client_list_lock);

	lpmemnode_t node = params->client_list, next = NULL;
	while (node != NULL) {
		next = node->next;
		struct ev_io *client = (struct ev_io *) MP_NODE_DATA(node);
		close(client->fd);
		mempool_free(g_mempool_client, node);
		node = next;
	}

	params->client_list = NULL;

	pthread_rwlock_unlock(&params->client_list_lock);
}

// --------------------------------------------------------

static void add_client_node(lpmemnode_t cli_node, server_params_t *params) {
	struct ev_io *client = (struct ev_io *) MP_NODE_DATA(cli_node);
	printf("add_client_node: %p\n", client);
	
	pthread_rwlock_wrlock(&params->client_list_lock);

	cli_node->next = params->client_list;
	params->client_list = cli_node;

	pthread_rwlock_unlock(&params->client_list_lock);
}

static void rmv_client_node(lpmemnode_t cli_node, server_params_t *params) {
	struct ev_io *client = (struct ev_io *) MP_NODE_DATA(cli_node);
	printf("rmv_client_node: %p\n", client);
	close(client->fd);
	
	pthread_rwlock_wrlock(&params->client_list_lock);

	lpmemnode_t node = params->client_list, prev_node = NULL;
	while (node != NULL) {
		if (node == cli_node) {
			if (prev_node == NULL) {
				params->client_list = node->next;
			} else {
				prev_node->next = node->next;
			}
			break;
		}
		prev_node = node;
		node = node->next;
	}

	pthread_rwlock_unlock(&params->client_list_lock);
}

// -----------------------------------------------------------

client_t * client_alloc(server_params_t *params) {
	lpmemnode_t mp_node = mempool_alloc(g_mempool_client);
	if (mp_node == NULL) {
		fprintf(stderr, "mempool_alloc client failed\n");
		return NULL;
	}
	client_t *data = (client_t *) MP_NODE_DATA(mp_node);
	data->params = params;
	add_client_node(mp_node, params);

	return data;
}

void client_return(client_t *data) {
	lpmemnode_t mp_node = MP_NODE_HEAD_PTR(data);
	rmv_client_node(mp_node, data->params);
	mempool_free(g_mempool_client, mp_node);
}

// -----------------------------------------------------------

char * packet_buffer_alloc() {
	lpmemnode_t mp_node = mempool_alloc(g_mempool_buffer);
	if (mp_node == NULL) {
		fprintf(stderr, "mempool_alloc buffer failed\n");
		return NULL;
	}
	return (char *) MP_NODE_DATA(mp_node);
}

void packet_buffer_return(char *data) {
	mempool_free(g_mempool_buffer, MP_NODE_HEAD_PTR(data));
}

// -----------------------------------------------------------

static send_req_t * send_req_alloc() {
	lpmemnode_t mp_node = mempool_alloc(g_mempool_send_req);
	if (mp_node == NULL) {
		fprintf(stderr, "mempool_alloc send_req failed\n");
		return NULL;
	}
	return (send_req_t *) MP_NODE_DATA(mp_node);
}

static void send_req_return(send_req_t *data) {
	mempool_free(g_mempool_send_req, MP_NODE_HEAD_PTR(data));
}

// -----------------------------------------------------------

static send_watcher_t * send_watcher_alloc() {
	lpmemnode_t mp_node = mempool_alloc(g_mempool_send_watcher);
	if (mp_node == NULL) {
		fprintf(stderr, "mempool_alloc send_watcher failed\n");
		return NULL;
	}
	return (send_watcher_t *) MP_NODE_DATA(mp_node);
}

static void send_watcher_return(send_watcher_t *data) {
	mempool_free(g_mempool_send_watcher, MP_NODE_HEAD_PTR(data));
}

// -----------------------------------------------------------

static void packet_send_cb(struct ev_loop *loop, struct ev_io *watcher, int revents) {
	send_watcher_t *send_watcher = (send_watcher_t *)watcher;
	if (EV_ERROR & revents) {
		perror("packet_send_cb got invalid event");
		return;
	}

	send_req_t *req = send_watcher->req;

	int send_num = send(watcher->fd, req->buf, req->buf_len, 0);
	if (send_num < 0) { // 有client关闭时
		perror("send error");
		// Stop and free watcher if client socket is closing
		ev_io_stop(loop, (struct ev_io *)send_watcher->tg_cli);
		client_return(send_watcher->tg_cli);
	}

	// printf("sent buf num: %d\n", send_num);

	req->rest_cnt--;
	if (req->rest_cnt == 0) {
		packet_buffer_return(req->buf);
		send_req_return(req);
	}

	ev_io_stop(loop, watcher);
	send_watcher_return(send_watcher);
}

int packet_broadcast(struct ev_loop *loop, client_t *src_cli, char *buf, ssize_t buf_len) {
	lpmemnode_t src_cli_node = MP_NODE_HEAD_PTR(src_cli);
	uint32_t target_ip = *(uint32_t *)buf;
	int sent = 0;
	char src_ip_str[20], tg_ip_str[20];
	server_params_t *params = src_cli->params;

	*(uint32_t *)buf = src_cli->ip; // 标注发出方的ip
	ip_to_str_ul(ntohl(src_cli->ip), src_ip_str);

	pthread_rwlock_rdlock(&params->client_list_lock);

	send_req_t *req = NULL;
	lpmemnode_t node = params->client_list;
	while (node != NULL) {
		client_t *tg_cli = (client_t *)MP_NODE_DATA(node);
		if ((target_ip == ALL_BROADCAST_IP && node != src_cli_node && tg_cli->ip != 0) || target_ip == tg_cli->ip) {
			if (req == NULL) {
				req = send_req_alloc();
				req->buf = buf;
				req->buf_len = buf_len;
				req->rest_cnt = 1;
			} else {
				req->rest_cnt++;
			}
		}
		node = node->next;
	}

	if (req != NULL) {
		node = params->client_list;
		while (node != NULL) {
			client_t *tg_cli = (client_t *)MP_NODE_DATA(node);
			if ((target_ip == ALL_BROADCAST_IP && node != src_cli_node && tg_cli->ip != 0) || target_ip == tg_cli->ip) {
				printf("broadcast_packet: (%s => %s), buf_len: %d\n"
					, src_ip_str, ip_to_str_ul(ntohl(tg_cli->ip), tg_ip_str), buf_len);

				send_watcher_t *send_watcher = send_watcher_alloc();
				if (send_watcher != NULL) {
					send_watcher->tg_cli = tg_cli;
					send_watcher->req = req;
					ev_io_init((struct ev_io *)send_watcher, packet_send_cb, tg_cli->watcher.fd, EV_WRITE);
					ev_io_start(loop, (struct ev_io *)send_watcher);
					
					sent = 1;
				}
			}
			node = node->next;
		}
	}

	pthread_rwlock_unlock(&params->client_list_lock);

	return sent;
}